package com.xiaohu.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo6GroupByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("groupByKey")

    val sc: SparkContext = new SparkContext(conf)

    //===================================================
    val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    //求每个班级的平均年龄
    val arrayRDD: RDD[Array[String]] = linesRDD.map((line: String) => line.split(","))


    //像这种RDD中的元素是(key,value)类型的，我们将这种RDD称之为键值对RDD(kv格式RDD)
    val clazzWithAgeRDD: RDD[(String, Int)] = arrayRDD.map {
      case Array(_, _, age: String, _, clazz: String) =>
        (clazz, age.toInt)
    }

    //    /**
    //     * groupBy算子的使用
    //     *
    //     * 1、groupBy的算子，后面的分组条件是我们自己指定的
    //     * 2、spark中groupBy之后的，所有值会被封装到一个Iterable迭代器中存储
    //     */
    //    // val map: Map[String, List[Score]] = scoreList.groupBy((s: Score) => s.id)
    //    val groupRDD: RDD[(String, Iterable[(String, Int)])] = clazzWithAgeRDD.groupBy(_._1)
    //    groupRDD.foreach(println)
    //
    //    val resKvRDD: RDD[(String, Double)] = groupRDD.map((kv: (String, Iterable[(String, Int)])) => {
    //      val clazz: String = kv._1
    //      val avgAge: Double = kv._2.map(_._2).sum.toDouble / kv._2.size
    //
    //      (clazz, avgAge)
    //    })
    //    resKvRDD.foreach(println)

    /**
     * GroupByKey属于kv格式的算子，只能作用在kv格式的RDD上
     * 也就说，只有kv格式的RDD才能调用kv格式的算子
     */
    val groupByKeyRDD: RDD[(String, Iterable[Int])] = clazzWithAgeRDD.groupByKey()

    val resKvRDD2: RDD[(String, Double)] = groupByKeyRDD.map((kv: (String, Iterable[Int])) => (kv._1, kv._2.sum.toDouble / kv._2.size))
    resKvRDD2.foreach(println)

    /**
     * 面试题：spark core中 groupBy算子与groupByKey算子的区别？
     * 1、代码格式上：
     * groupBy的分组条件可以自己指定，并且绝大部分的RDD都可以调用该算子，返回的是键和元素本身组成的迭代器构成的kv格式RDD
     * groupByKey算子，只能由kv格式的RDD进行调用，分组的条件会自动根据键进行分组，不需要在自己指定，返回的是键和值组成的迭代器构成的kv格式RDD
     *
     * 2、执行shuffle数据量来看
     *  groupBy产生的shuffle数据量在一定程度上要大于groupByKey产生的shuffle数据量
     *  所以groupByKey算子的执行效率要比groupBy算子的执行效率要高
     */

    while (true) {

    }


  }
}
